package io.github.wanggit.antrpc.client.monitor;

import io.github.wanggit.antrpc.client.monitor.report.*;
import io.github.wanggit.antrpc.commons.bean.RpcCallLog;
import io.github.wanggit.antrpc.commons.config.CallLogReporterConfig;
import io.github.wanggit.antrpc.commons.config.IConfiguration;
import io.vavr.Function3;
import lombok.extern.slf4j.Slf4j;
import org.springframework.context.ApplicationContext;

import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.concurrent.ArrayBlockingQueue;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;

/** 日志上报工具 */
@Slf4j
public class RpcCallLogHolder implements IRpcCallLogHolder {

    private final List<ILogReporter> logReporters = new ArrayList<>();

    private final ThreadPoolExecutor threadPoolExecutor;

    private final Map<String, List<IRpcCallLogListener>> listeners = new HashMap<>();

    private final Function3<
                    ZipkinAllMethodTracing.TracingPairs, RpcCallLog, List<ILogReporter>, Void>
            function3 =
                    (lastTracingPairs, rpcCallLog, logReporters) -> {
                        fireListener(rpcCallLog);
                        for (ILogReporter logReporter : logReporters) {
                            logReporter.report(rpcCallLog, lastTracingPairs);
                        }
                        return null;
                    };

    public RpcCallLogHolder(IConfiguration configuration, ApplicationContext applicationContext) {
        CallLogReporterConfig logHolderConfig = configuration.getCallLogReporterConfig();
        reInitReporters(logHolderConfig, configuration, applicationContext);
        threadPoolExecutor =
                new ThreadPoolExecutor(
                        3,
                        6,
                        2,
                        TimeUnit.SECONDS,
                        new ArrayBlockingQueue<>(10),
                        (r, executor) -> {
                            if (log.isWarnEnabled()) {
                                log.warn("The log queue is full and some logs will be discarded.");
                            }
                        });
    }

    @Override
    public void reInitReporters(
            CallLogReporterConfig logHolderConfig,
            IConfiguration configuration,
            ApplicationContext applicationContext) {
        logReporters.clear();
        if (logHolderConfig.isReportToKafka()) {
            logReporters.add(new KafkaLogReporter(configuration, applicationContext));
        }
        if (logHolderConfig.isReportToSlf4j()) {
            logReporters.add(new Slf4jInfoLogReporter(configuration, applicationContext));
        }
        if (logHolderConfig.isReportToZipkin()) {
            logReporters.add(new ZipkinReporter(configuration, applicationContext));
        }
    }

    @Override
    public void log(RpcCallLog rpcCallLog) {
        threadPoolExecutor.submit(
                new CallLogRunnableTask(
                        ZipkinAllMethodTracing.getLastTracingPairs(),
                        rpcCallLog,
                        logReporters,
                        function3));
    }

    private void fireListener(RpcCallLog rpcCallLog) {
        String key = rpcCallLog.getTargetClassName() + "#" + rpcCallLog.getTargetMethodName();
        List<IRpcCallLogListener> iRpcCallLogListeners = listeners.get(key);
        if (null != iRpcCallLogListeners) {
            for (IRpcCallLogListener rpcCallLogListener : iRpcCallLogListeners) {
                rpcCallLogListener.listen(rpcCallLog);
            }
        }
    }

    @Override
    public void addListener(String name, IRpcCallLogListener rpcCallLogListener) {
        if (!listeners.containsKey(name)) {
            listeners.put(name, new ArrayList<>());
        }
        listeners.get(name).add(rpcCallLogListener);
    }

    @Override
    public void removeListener(String name, IRpcCallLogListener rpcCallLogListener) {
        List<IRpcCallLogListener> iRpcCallLogListeners = listeners.get(name);
        if (null != iRpcCallLogListeners) {
            iRpcCallLogListeners.remove(rpcCallLogListener);
        }
    }

    public static class CallLogRunnableTask implements Runnable {

        private final ZipkinAllMethodTracing.TracingPairs lastTracingPairs;
        private final Function3<
                        ZipkinAllMethodTracing.TracingPairs, RpcCallLog, List<ILogReporter>, Void>
                function3;
        private final RpcCallLog rpcCallLog;
        private final List<ILogReporter> logReporters;

        CallLogRunnableTask(
                ZipkinAllMethodTracing.TracingPairs lastTracingPairs,
                RpcCallLog rpcCallLog,
                List<ILogReporter> logReporters,
                Function3<ZipkinAllMethodTracing.TracingPairs, RpcCallLog, List<ILogReporter>, Void>
                        function3) {
            this.lastTracingPairs = lastTracingPairs;
            this.function3 = function3;
            this.rpcCallLog = rpcCallLog;
            this.logReporters = logReporters;
        }

        @Override
        public void run() {
            function3.apply(lastTracingPairs, rpcCallLog, logReporters);
        }
    }
}
